package org.renhj.consumer;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import org.renhj.common.RpcRequest;
import org.renhj.common.RpcResponse;
import org.renhj.handler.RpcConsumerHandler;
import org.renhj.metric.ZookeeperMetric;
import org.renhj.rpc.RpcDecoder;
import org.renhj.rpc.RpcEncoder;

import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

public class RpcClient {

    private String host;
    private int port;

    private Channel channel;
    private final NioEventLoopGroup group = new NioEventLoopGroup();

    public RpcClient() {
    }

    public RpcClient(String address) {
        this.host = address.split(":")[0];
        this.port = Integer.parseInt(address.split(":")[1]);
    }

    public RpcClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void initConnect() throws InterruptedException {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast(new IdleStateHandler(0,0, 5, TimeUnit.SECONDS));
                        pipeline.addLast(new RpcEncoder(RpcRequest.class));
                        pipeline.addLast(new RpcDecoder(RpcResponse.class));
                        pipeline.addLast(new RpcConsumerHandler());
                    }
                });


        this.channel = bootstrap.connect(this.host, this.port).sync().channel();

        if (!isValidate()) {
            this.close();
            return;
        }

        System.out.println("启动客户端： " + host + ":" + port);
    }


    private boolean isValidate() {
        if (this.channel != null) {
            return this.channel.isActive();
        }
        return false;
    }

    public void close() {
        if (this.channel != null && this.channel.isActive()) {
            this.channel.close();
        }
        if (!this.group.isShutdown()) {
            this.group.shutdownGracefully();
        }
        System.out.println("rpc client [" + host + ":" + port + "] close");
    }

    public Object send(RpcRequest request) throws Exception {
        ZookeeperMetric.getInstance().createMetricNode(request.getRequestId(), host + ":" + port);
        return this.channel.writeAndFlush(request).sync().get();
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        RpcClient rpcClient = (RpcClient) o;
        return port == rpcClient.port && Objects.equals(host, rpcClient.host);
    }

    @Override
    public int hashCode() {
        return Objects.hash(host, port);
    }


    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }
}
